package com.poetic.network.flow.function;

import com.poetic.network.flow.domain.UserBehavior;
import com.poetic.network.flow.domain.UvCount;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Set;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-15.
 * </pre>
 *
 * @author lianghuikun
 */
public class UVCountByWindow implements AllWindowFunction<UserBehavior, UvCount, TimeWindow> {
    @Override
    public void apply(TimeWindow window, Iterable<UserBehavior> input, Collector<UvCount> out) throws Exception {

        Set<Long> idSet = new HashSet<>();

        input.forEach(userBehavior -> idSet.add(userBehavior.getUserId()));

        UvCount uvCount = UvCount.builder()
                .windowEnd(window.getEnd())
                .count(idSet.size())
                .build();

        out.collect(uvCount);
    }
}
